package org.apache.rocketmq.broker;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.utils.NetworkUtil;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.netty.NettyClientConfig;
import org.apache.rocketmq.remoting.netty.NettyServerConfig;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.srvutil.ServerUtil;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.config.MessageStoreConfig;

import java.io.BufferedInputStream;
import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicInteger;



/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/1/8
 * @方法描述：Broker模块的启动类
 */
public class BrokerStartup {

    public static Logger log;

    //系统配置文件加载器
    public static final SystemConfigFileHelper CONFIG_FILE_HELPER = new SystemConfigFileHelper();


    public static void main(String[] args) {
        //在start方法中，会创建BrokerController对象，然后启动BrokerController
        start(createBrokerController(args));
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/8
     * @方法描述：启动Broker程序的入口方法
     */
    public static BrokerController start(BrokerController controller) {
        try {
            //启动BrokerController
            controller.start();

            //得到启动的Broker的名称，地址，以及序列化类型
            String tip = String.format("The broker[%s, %s] boot success. serializeType=%s",
                    controller.getBrokerConfig().getBrokerName(), controller.getBrokerAddr(),
                    RemotingCommand.getSerializeTypeConfigInThisServer());

            //如果Broker的配置信息中包含Namesrv地址，就把这个地址也拼接到刚才得到的字符串中
            if (null != controller.getBrokerConfig().getNamesrvAddr()) {
                tip += " and name server is " + controller.getBrokerConfig().getNamesrvAddr();
            }

            //记录日志信息
            log.info(tip);
            //在控制台输出字符串信息
            System.out.printf("%s%n", tip);
            return controller;
        } catch (Throwable e) {
            e.printStackTrace();
            System.exit(-1);
        }
        return null;
    }

    //停止Broker程序工作的方法
    public static void shutdown(final BrokerController controller) {
        if (null != controller) {
            controller.shutdown();
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/8
     * @方法描述：创建BrokerController对象的方法
     */
    public static BrokerController buildBrokerController(String[] args) throws Exception {
        //把Rocketmq协议的版本号设置到系统属性中
        System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
        //创建BrokerConfig配置信息对象
        final BrokerConfig brokerConfig = new BrokerConfig();
        //创建NettyServerConfig配置信息对象，在第五版本代码中，虽然我为Broker创建并启动了Broker模块的Netty服务端
        //但是这个服务端并没有用到，等到第六版本才会真正用到Netty的服务端
        final NettyServerConfig nettyServerConfig = new NettyServerConfig();
        //创建NettyClientConfig配置信息对象，在第五版本代码中，会使用Netty构建的客户端，定期为Broker模块向NameServer发送心跳消息
        final NettyClientConfig nettyClientConfig = new NettyClientConfig();
        final MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
        //设置Broker服务器监听的端口号
        nettyServerConfig.setListenPort(10911);
        //设置消息存储引擎的HA监听端口，这里设置为0，表示不启用HA监听
        messageStoreConfig.setHaListenPort(0);

        //创建一个Options对象，用于存储用户配置的命令行参数信息
        Options options = ServerUtil.buildCommandlineOptions(new Options());
        //解析命令行参数信息，这些代码在前面几个版本应该都见过了，我就不再重复解释了
        CommandLine commandLine = ServerUtil.parseCmdLine("mqbroker", args, buildCommandlineOptions(options), new DefaultParser());
        //命令行参数解析失败则退出程序
        if (null == commandLine) {
            System.exit(-1);
        }

        //定义一个Properties对象，用来存储配置文件中的数据信息
        Properties properties = null;
        //判断命令行参数信息中是否包含配置信息路径的命令，在我为大家提供的第五版本代码测试类中，就在命令行中定义了配置文件的路径
        if (commandLine.hasOption('c')) {
            //得到配置文件路径
            String file = commandLine.getOptionValue('c');
            if (file != null) {
                //把配置文件路径交给系统配置文件加载器
                CONFIG_FILE_HELPER.setFile(file);
                //注意，这里还做了一个操作，这里把用户定义的配置文件的路径也设置到BrokerPathConfigHelper对象中了
                //后面会使用这个BrokerPathConfigHelper对象获得配置文件路径
                BrokerPathConfigHelper.setBrokerConfigPath(file);
                //加载配置文件中的数据到Properties对象中
                properties = CONFIG_FILE_HELPER.loadConfig();
            }
        }

        //判断Properties对象是否为空
        if (properties != null) {
            //不为空则把Properties对象中的属性设置到系统环境变量中
            properties2SystemEnv(properties);
            //把Properties对象中的属性设置到BrokerConfig对象中
            MixAll.properties2Object(properties, brokerConfig);
            //把Properties对象中的属性设置到NettyServerConfig对象中
            MixAll.properties2Object(properties, nettyServerConfig);
            //把Properties对象中的属性设置到NettyClientConfig对象中
            MixAll.properties2Object(properties, nettyClientConfig);
        }

        //把用户设置的命令行参数信息设置到BrokerConfig对象中，如果用户把Namesrv的地址设置到命令行参数中了
        //那这个时候就会把这个地址设置到BrokerConfig对象中
        MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), brokerConfig);

        //判断BrokerConfig对象中的RocketMQ路径是未被设置
        if (null == brokerConfig.getRocketmqHome()) {
            System.out.printf("Please set the %s variable in your environment " +
                    "to match the location of the RocketMQ installation", MixAll.ROCKETMQ_HOME_ENV);
            //没有被设置则直接退出程序
            System.exit(-2);
        }

        //得到BrokerConfig对象中的Namesrv地址
        String namesrvAddr = brokerConfig.getNamesrvAddr();
        //对Namesrv地址判空
        if (StringUtils.isNotBlank(namesrvAddr)) {
            try {
                //这里对Namesrv地址进行了分割，因为用户可以在命令行中设置多个Namesrv地址，Namesrv本身就可以以集群模式来构建
                //但是在我们的代码中，目前构建的都是单机模式
                String[] addrArray = namesrvAddr.split(";");
                //校验每一个Namesrv地址是否合法
                for (String addr : addrArray) {
                    NetworkUtil.string2SocketAddress(addr);
                }
            } catch (Exception e) {
                System.out.printf("The Name Server Address[%s] illegal, please set it as follows, " +
                        "\"127.0.0.1:9876;192.168.0.1:9876\"%n", namesrvAddr);
                //地址格式错误则直接退出程序
                System.exit(-3);
            }
        }

        //如果Broker的角色配置为SLAVE从节点
        if (BrokerRole.SLAVE == messageStoreConfig.getBrokerRole()) {
            //获取当前配置的内存访问消息比率，并减少10
            int ratio = messageStoreConfig.getAccessMessageInMemoryMaxRatio() - 10;
            //更新内存访问消息比率设置，这是为了优化从节点的内存使用
            messageStoreConfig.setAccessMessageInMemoryMaxRatio(ratio);
        }


        if (messageStoreConfig.getHaListenPort() <= 0) {
            messageStoreConfig.setHaListenPort(nettyServerConfig.getListenPort() + 1);
        }

        //设置当前Broker不在容器中运行的标志
        brokerConfig.setInBrokerContainer(false);
        //设置Broker模块中存储日志信息的文件夹路径
        System.setProperty("brokerLogDir", "");

        //判断是否启用了日志隔离功能，如果一台服务器上部署了多个Broker，那么就应该启动日志隔离功能
        if (brokerConfig.isIsolateLogEnable()) {
            //记录日志的文件夹路径就被定义为各个Broker的名称+Broker的ID
            System.setProperty("brokerLogDir", brokerConfig.getBrokerName() + "_" + brokerConfig.getBrokerId());
        }


        //判断命令行是否包含'p'命令，如果包含，则把配置信息输出到控制台
        if (commandLine.hasOption('p')) {
            Logger console = LoggerFactory.getLogger(LoggerName.BROKER_CONSOLE_NAME);
            MixAll.printObjectProperties(console, brokerConfig);
            MixAll.printObjectProperties(console, nettyServerConfig);
            MixAll.printObjectProperties(console, nettyClientConfig);
            MixAll.printObjectProperties(console, messageStoreConfig);
            //输出完之后退出程序
            System.exit(0);
        } else if (commandLine.hasOption('m')) {
            //如果命令行包含'm'命令，则以掩码方式输出Broker配置的所有信息到控制台，敏感信息将被掩码处理
            Logger console = LoggerFactory.getLogger(LoggerName.BROKER_CONSOLE_NAME);
            MixAll.printObjectProperties(console, brokerConfig, true);
            MixAll.printObjectProperties(console, nettyServerConfig, true);
            MixAll.printObjectProperties(console, nettyClientConfig, true);
            MixAll.printObjectProperties(console, messageStoreConfig, true);
            //输出完之后退出程序
            System.exit(0);
        }

        //得到记录Broker日志信息的Logger对象
        log = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
        //记录brokerConfig对象日志信息
        MixAll.printObjectProperties(log, brokerConfig);
        //记录nettyServerConfig对象日志信息
        MixAll.printObjectProperties(log, nettyServerConfig);
        //记录nettyClientConfig对象日志信息
        MixAll.printObjectProperties(log, nettyClientConfig);
        MixAll.printObjectProperties(log, messageStoreConfig);
        //在这里创建了BrokerController对象
        final BrokerController controller = new BrokerController(brokerConfig, nettyServerConfig, nettyClientConfig,messageStoreConfig);
        //把用户定义在配置文件中的内容交给BrokerController对象的配置管理器管理了
        controller.getConfiguration().registerConfig(properties);
        return controller;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/8
     * @方法描述：创建Broker模块终止运行时的钩子方法，该方法会交给JVM，在JVM关闭的时候该方法会被执行
     */
    public static Runnable buildShutdownHook(BrokerController brokerController) {
        return new Runnable() {
            //为当前任务定义一个判断程序是否已经关闭的标志
            private volatile boolean hasShutdown = false;
            //定义一个计数器，记录当前任务被执行的次数，其实就代表Broker被关闭的次数
            //这个是用来排查Broker异常关闭问题的
            private final AtomicInteger shutdownTimes = new AtomicInteger(0);

            @Override
            public void run() {
                synchronized (this) {
                    //记录程序关闭的日志信息
                    log.info("Shutdown hook was invoked, {}", this.shutdownTimes.incrementAndGet());
                    //判断程序是否已经关闭
                    if (!this.hasShutdown) {
                        //没有关闭则设置程序已经关闭的标志
                        this.hasShutdown = true;
                        //得到程序开始关闭的时间
                        long beginTime = System.currentTimeMillis();
                        //执行真正关闭Broker程序地操作
                        brokerController.shutdown();
                        //计算关闭程序消耗的时间
                        long consumingTimeTotal = System.currentTimeMillis() - beginTime;
                        //记录日志信息
                        log.info("Shutdown hook over, consuming total time(ms): {}", consumingTimeTotal);
                    }
                }
            }
        };
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/8
     * @方法描述：创建BrokerController对象的方法
     */
    public static BrokerController createBrokerController(String[] args) {
        try {
            //创建BrokerController对象
            BrokerController controller = buildBrokerController(args);
            //初始化BrokerController对象，在这个过程中会把Broker模块用到的组件都创建并且初始化完毕
            boolean initResult = controller.initialize();
            if (!initResult) {
                //初始化失败的话则退出程序
                controller.shutdown();
                System.exit(-3);
            }
            //把构建好的Broker关闭时的钩子方法交给JVM，这样一来，当JVM关闭的时候，这个方法就会被执行
            Runtime.getRuntime().addShutdownHook(new Thread(buildShutdownHook(controller)));
            return controller;
        } catch (Throwable e) {
            e.printStackTrace();
            System.exit(-1);
        }
        return null;
    }



    //设置获取Namesrv域名和子组的方法
    private static void properties2SystemEnv(Properties properties) {
        if (properties == null) {
            return;
        }
        String rmqAddressServerDomain = properties.getProperty("rmqAddressServerDomain", MixAll.WS_DOMAIN_NAME);
        String rmqAddressServerSubGroup = properties.getProperty("rmqAddressServerSubGroup", MixAll.WS_DOMAIN_SUBGROUP);
        //设置的这些属性在DefaultTopAddressing对象中会用到，大家可以直接去DefaultTopAddressing类中查看详细注释
        //我在DefaultTopAddressing类中解释了为什么要设置Namesrv的域名和子组
        System.setProperty("rocketmq.namesrv.domain", rmqAddressServerDomain);
        System.setProperty("rocketmq.namesrv.domain.subgroup", rmqAddressServerSubGroup);
    }

    //构建封装命令行参数的Options对象的方法
    private static Options buildCommandlineOptions(final Options options) {
        Option opt = new Option("c", "configFile", true, "Broker config properties file");
        opt.setRequired(false);
        options.addOption(opt);

        opt = new Option("p", "printConfigItem", false, "Print all config item");
        opt.setRequired(false);
        options.addOption(opt);

        opt = new Option("m", "printImportantConfig", false, "Print important config item");
        opt.setRequired(false);
        options.addOption(opt);
        return options;
    }

    //这个内部类的对象就是系统文件加载器
    public static class SystemConfigFileHelper {

        private static final Logger LOGGER = LoggerFactory.getLogger(SystemConfigFileHelper.class);

        //用户定义的配置文件的路径会赋值给这个成员变量
        private String file;

        public SystemConfigFileHelper() {
        }

        //加载配置文件到Properties对象中的方法
        public Properties loadConfig() throws Exception {
            InputStream in = new BufferedInputStream(Files.newInputStream(Paths.get(file)));
            Properties properties = new Properties();
            properties.load(in);
            in.close();
            return properties;
        }


        public void update(Properties properties) throws Exception {
            LOGGER.error("[SystemConfigFileHelper] update no thing.");
        }


        public void setFile(String file) {
            this.file = file;
        }

        public String getFile() {
            return file;
        }
    }
}
